const { Transform } = require("stream");
const _ = require("lodash");
const utils = require('../../utils');

function querySQL(db, sql) {
  return new Promise(resolve => {
    utils.debug(`sql: ${sql}`);
    db.query(sql, (err, results) => {
      utils.debug('querySQL:', { err, results });
      resolve({ err, results });
    });
  })
}

// fields： { phone: 'VARCHAR(11)', password: 'VARCHAR(50)' }
module.exports = async function ({ user, password, connectString, table, fields }) {
  const stream = new Transform({ objectMode: true });
  const oracledb = require("oracledb");
  const db = await oracledb.getConnection({ user, password, connectString });
  db.connect();
  let hasTable = !!fields;

  if (fields) {
    fields.id = `int(11) NOT NULL AUTO_INCREMENT`;
    const _fields = _.map(fields, (o, k) => `${k} ${o}`).join(',');
    await querySQL(db, `CREATE TABLE IF NOT EXISTS ${table} (${_fields}, PRIMARY KEY (id) USING BTREE) AUTO_INCREMENT = 1 CHARACTER SET = utf8 COLLATE = utf8_general_ci`);
    hasTable = true;
  }

  stream._transform = async function (chunk, encoding, callback) {
    utils.debug('oracleWriter [chunk]:', chunk);
    stream.push(chunk);

    await utils.until(() => hasTable);
    const cols = _.keys(chunk);
    const values = cols.map(o => `'${chunk[o]}'`);
    await querySQL(db, `INSERT INTO ${table} (${cols.join(',')}) VALUES (${values.join(',')})`);

    callback();
  };

  stream.on('finish', async function () {
    db.close();
    utils.debug('oracleWriter [finish]');
  });

  return stream;
};
